package com.radica.service.external.distribution.customer.impl;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.nio.charset.Charset;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Set;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.commons.validator.routines.DoubleValidator;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import com.google.code.externalsorting.ExternalSort;
import com.radica.converter.DBObjectConverter;
import com.radica.db.entity.customer.CustomerDBVO;
import com.radica.db.entity.customer.CustomerPropertyDBVO;
import com.radica.db.entity.customer.CustomerPropertyItemDBVO;
import com.radica.db.entity.customer.EmailSubscriberDBVO;
import com.radica.db.entity.customer.MobileSubscriberDBVO;
import com.radica.db.entity.system.email.EmailNotificationDBVO;
import com.radica.db.valueobject.schedule.context.CustomerMappingColumnVO;
import com.radica.db.valueobject.schedule.context.CustomerUploadResultVO;
import com.radica.db.valueobject.schedule.context.CustomerUploadVO;
import com.radica.enums.customer.CustomerColumnType;
import com.radica.enums.customer.CustomerPropertyType;
import com.radica.repository.mongo.system.email.EmailNotificationAutoRepo;
import com.radica.service.internal.constant.InternalConstants;
import com.radica.service.internal.constant.RedisKeyName;
import com.radica.util.CsvWriter;
import com.radica.util.CustomerUploadFileUtil;
import com.radica.util.EmailUtil;
import com.radica.util.ObjectIdGenerator;
import com.radica.util.PhoneNumberUtil;
import com.radica.util.PhoneNumberUtil.PhoneNumber;
import com.radica.util.RICsvReader;
import com.radica.util.RIExcelReader;
import com.radica.util.RIReader;
import com.radica.util.exception.InvalidPhoneNumberException;

@Service
public class CustomerUploadHelper {

	private static final Logger logger = LogManager.getLogger(CustomerUploadHelper.class);

	private static long splitNumber = 3;
	private static long splitThread = 3;

	static String MANGGISID_NOT_NULL_FILE_NAME_SUFFIX = "_manggisIdNotNull.csv";
	static String EXTID_NOT_NULL_FILE_NAME_SUFFIX = "_extIdNotNull.csv";
	static String EMAIL_NOT_NULL_FILE_NAME_SUFFIX = "_emailNotNull.csv";
	static String MOBILE_NOT_NULL_FILE_NAME_SUFFIX = "_mobileNotNull.csv";

	static String SORT_MANGGISID_NOT_NULL_FILE_NAME_SUFFIX = "_sortManggisIdNotNull.csv";
	static String SORT_EXTID_NOT_NULL_FILE_NAME_SUFFIX = "_sortExtIdNotNull.csv";
	static String SORT_EMAIL_NOT_NULL_FILE_NAME_SUFFIX = "_sortEmailNotNull.csv";
	static String SORT_MOBILE_NOT_NULL_FILE_NAME_SUFFIX = "_sortMobileNotNull.csv";

	private final String SET_SEPARATOR = "\\|\\|";
	@Value("${property_resource_path}")
	private String resourcePath;
	@Value("${property_customer_date_format}")
	private String customerDateFormat;
	@Autowired
	private DBObjectConverter dbObjectConverter;
	@Autowired
	private EmailNotificationAutoRepo emailNotificationAutoRepo;
	@Autowired
	private JedisPool jedisPool;
	static long batchSize_updateRedis = 5000;

	/**
	 * 根据指定的四个匹配property(manggisId, externalId, email, mobile)中的一个,
	 * 检查文件里是否存在对应的column 如果这个文件存在指定的property的column, 则return true. 否则return false.
	 * 
	 * @param matchedMap
	 * @param propertyType
	 * @return
	 */
	boolean checkUniqueKeyColumnExist(CustomerUploadVO context, CustomerPropertyType propertyType) {
		boolean exitColumn = false;
		if (CustomerPropertyType.MANGGISID == propertyType) {
			if (context.getManggisIdColumnIndex() != -1) {
				exitColumn = true;
			}
		} else if (CustomerPropertyType.EXTERNALID == propertyType) {
			if (context.getExternalIdColumnIndex() != -1) {
				exitColumn = true;
			}
		} else if (CustomerPropertyType.EMAIL == propertyType) {
			if (context.getEmailColumnIndex() != -1) {
				exitColumn = true;
			}
		} else if (CustomerPropertyType.MOBILE == propertyType) {
			if (context.getMobileColumnIndex() != -1) {
				exitColumn = true;
			}
		}
		return exitColumn;
	}

	private String populateOriginalFilePath(String fileName) {
		StringBuilder originalFilePathBuilder = new StringBuilder(200);
		originalFilePathBuilder.append(resourcePath);
		originalFilePathBuilder.append(InternalConstants.UPLOAD_CUSTOMER_ORIGINAL);
		originalFilePathBuilder.append(fileName);
		String originalFilePath = originalFilePathBuilder.toString();
		return originalFilePath;
	}

	// 验证数据, 把源文件中没有问题的数据拆分为manggisId, extId, email,
	// mobile和所有key都为空的相互独立的文件。然后update resultContext中的totalRecord
	String[] validateExportToCsvFileAndUpdateTotalRecord(CustomerUploadVO context, CustomerUploadResultVO resultContext, String jobId,
			Set<CustomerPropertyDBVO> properties, String listId, WrapBufferedWriter wrapInsertBufferedWriter) throws Exception {
		String[] generateCsvHeader = null;
		boolean existManggisIdColumn = checkUniqueKeyColumnExist(context, CustomerPropertyType.MANGGISID);
		boolean existExternalIdColumn = checkUniqueKeyColumnExist(context, CustomerPropertyType.EXTERNALID);
		boolean existEmailColumn = checkUniqueKeyColumnExist(context, CustomerPropertyType.EMAIL);
		boolean existMobileColumn = checkUniqueKeyColumnExist(context, CustomerPropertyType.MOBILE);

		String uploadCustomerDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER;

		String manggisIdNotNullPath = uploadCustomerDirectory + jobId + MANGGISID_NOT_NULL_FILE_NAME_SUFFIX;
		String extIdNotNullPath = uploadCustomerDirectory + jobId + EXTID_NOT_NULL_FILE_NAME_SUFFIX;
		String emailNotNullPath = uploadCustomerDirectory + jobId + EMAIL_NOT_NULL_FILE_NAME_SUFFIX;
		String mobileNotNullPath = uploadCustomerDirectory + jobId + MOBILE_NOT_NULL_FILE_NAME_SUFFIX;

		String originalFilePath = populateOriginalFilePath(context.getFileName());
		RIReader originalFileReader = null;
		RIReader calculatorReader = null;
		RIReader splitReader = null;
		WrapWriter fieldProblemWrapWriter = new WrapWriter();
		WrapWriter manggisIdNotFoundWrapWriter = new WrapWriter();
		WrapWriter firstNameLastNameEmptyWrapWriter = new WrapWriter();

		CsvWriter manggisIdNotNullWriter = null;
		CsvWriter extIdNotNullWriter = null;
		CsvWriter emailNotNullWriter = null;
		CsvWriter mobileNotNullWriter = null;
		List<CsvWriter> splitFileWriterList = new ArrayList<CsvWriter>();

		Jedis jedis = jedisPool.getResource();
		try {
			boolean isCsv = StringUtils.endsWithIgnoreCase(originalFilePath, ".csv");
			if (isCsv) {
				originalFileReader = new RICsvReader(originalFilePath);
				calculatorReader = new RICsvReader(originalFilePath);
				splitReader = new RICsvReader(originalFilePath);
			} else {
				originalFileReader = new RIExcelReader(originalFilePath);
				calculatorReader = new RIExcelReader(originalFilePath);
			}
			generateCsvHeader = CustomerUploadFileUtil.extractHeader(context.getMatchColumn(), originalFileReader.getHeaderCount());
			long totalRecord = 0;
			long issueRecord = 0;
			while (calculatorReader.hasNext()) {
				totalRecord++;
			}
			// refector start
			long fullRecords = totalRecord / (splitNumber * splitThread);
			long remainRecords = totalRecord - (splitNumber * splitThread) * fullRecords;

			long everyRemain = remainRecords / splitThread;
			long temp = remainRecords % splitThread;

			long fullNumberCount = fullRecords * splitThread;
			long splitTotalCount = 0;

			for (int i = 0; i < fullNumberCount; i++) {
				CsvWriter splitWriter = null;
				splitFileWriterList.add(splitWriter);
			}

			for (int i = 0; i < fullNumberCount; i++) {
				if (splitFileWriterList.get(i) == null) {
					CsvWriter splitWriter = new CsvWriter(manggisIdNotNullPath, Charset.defaultCharset());
					splitWriter.writeRecord(generateCsvHeader);
					splitFileWriterList.add(splitWriter);
				}
				while (splitTotalCount % splitNumber != 0) {
					splitFileWriterList.get(i).writeRecord(originalFileReader.getColumnValues());
					splitTotalCount++;
				}
			}
			System.out.println("remainCount:" + remainRecords);

			if (everyRemain != 0) {
				for (int i = 0; i < splitThread - 1; i++) {
					System.out.println(everyRemain);
				}
			}
			if (everyRemain + temp != 0) {
				System.out.println(everyRemain + temp);
			}
			// refector end

			resultContext.setTotalRecord(totalRecord);
			jedis.set(jobId + RedisKeyName.CUSTOMER_TOTAL_CUSTOMER_UPLOAD_COUNT, totalRecord + "");
			while (originalFileReader.hasNext()) {
				boolean existIssue = validateIssue(context, resultContext, originalFileReader.getColumnValues(), properties, jobId, generateCsvHeader, listId,
						fieldProblemWrapWriter, manggisIdNotFoundWrapWriter, firstNameLastNameEmptyWrapWriter);
				if (!existIssue) {
					String manggisId = null;
					String externalId = null;
					String email = null;
					String mobile = null;
					if (existManggisIdColumn) {
						manggisId = StringUtils.strip(originalFileReader.getStringColumnValue(context.getManggisIdColumnIndex()));
					}
					if (existExternalIdColumn) {
						externalId = StringUtils.strip(originalFileReader.getStringColumnValue(context.getExternalIdColumnIndex()));
					}
					if (existEmailColumn) {
						email = StringUtils.strip(originalFileReader.getStringColumnValue(context.getEmailColumnIndex()));
					}
					if (existMobileColumn) {
						mobile = StringUtils.strip(originalFileReader.getStringColumnValue(context.getMobileColumnIndex()));
					}

					if (StringUtils.isNotEmpty(manggisId)) {
						if (manggisIdNotNullWriter == null) {
							manggisIdNotNullWriter = new CsvWriter(manggisIdNotNullPath, Charset.defaultCharset());
							manggisIdNotNullWriter.writeRecord(generateCsvHeader);
						}
						manggisIdNotNullWriter.writeRecord(originalFileReader.getColumnValues());
					} else if (StringUtils.isNotEmpty(externalId)) {
						if (extIdNotNullWriter == null) {
							extIdNotNullWriter = new CsvWriter(extIdNotNullPath, Charset.defaultCharset());
							extIdNotNullWriter.writeRecord(generateCsvHeader);
						}
						extIdNotNullWriter.writeRecord(originalFileReader.getColumnValues());
					} else if (StringUtils.isNotEmpty(email)) {
						if (emailNotNullWriter == null) {
							emailNotNullWriter = new CsvWriter(emailNotNullPath, Charset.defaultCharset());
							emailNotNullWriter.writeRecord(generateCsvHeader);
						}
						emailNotNullWriter.writeRecord(originalFileReader.getColumnValues());
					} else if (StringUtils.isNotEmpty(mobile)) {
						if (mobileNotNullWriter == null) {
							mobileNotNullWriter = new CsvWriter(mobileNotNullPath, Charset.defaultCharset());
							mobileNotNullWriter.writeRecord(generateCsvHeader);
						}
						mobileNotNullWriter.writeRecord(originalFileReader.getColumnValues());
					} else {
						CustomerDBVO customer = convertColumnValueToCustomerDBVO(context, properties, originalFileReader.getColumnValues());
						this.writeColumnValueToInsertPrepareFile(customer, listId, context.getActionSetIds(), jobId, wrapInsertBufferedWriter);
						resultContext.setInsertNumber(resultContext.getInsertNumber() + 1);
					}
				} else {
					issueRecord++;
					jedisIncrByCount(jedis, jobId, issueRecord);
				}
			}
			jedisIncrByRemain(jedis, jobId, issueRecord);
		} catch (Exception e) {
			logger.error("validateExportToCsvFileAndUpdateTotalRecord Caught exception:" + ExceptionUtils.getStackTrace(e));
			throw e;
		} finally {
			jedisPool.returnResource(jedis);
			if (calculatorReader != null) {
				calculatorReader.close();
			}
			if (originalFileReader != null) {
				originalFileReader.close();
			}
			if (fieldProblemWrapWriter.getCsvWriter() != null) {
				fieldProblemWrapWriter.getCsvWriter().close();
			}
			if (manggisIdNotFoundWrapWriter.getCsvWriter() != null) {
				manggisIdNotFoundWrapWriter.getCsvWriter().close();
			}
			if (firstNameLastNameEmptyWrapWriter.getCsvWriter() != null) {
				firstNameLastNameEmptyWrapWriter.getCsvWriter().close();
			}
			if (manggisIdNotNullWriter != null) {
				manggisIdNotNullWriter.close();
			}
			if (extIdNotNullWriter != null) {
				extIdNotNullWriter.close();
			}
			if (emailNotNullWriter != null) {
				emailNotNullWriter.close();
			}
			if (mobileNotNullWriter != null) {
				mobileNotNullWriter.close();
			}
		}
		return generateCsvHeader;
	}
	void writeColumnValueToInsertPrepareFile(CustomerDBVO customerDBVO, String listId, Set<String> actionSetIds, String jobId,
			WrapBufferedWriter wrapInsertBufferedWriter) throws IOException {
		Date date = new Date();
		customerDBVO.setId(ObjectIdGenerator.generate());
		customerDBVO.setCreateTime(date);
		customerDBVO.setUpdateTime(date);
		customerDBVO.setCustomerListId(listId);
		customerDBVO.setActionSetIds(actionSetIds);
		if (wrapInsertBufferedWriter.getBufferedWriter() == null) {
			String insertFilePath = resourcePath + InternalConstants.UPLOAD_CUSTOMER_PREPARE_BG_INSERT + jobId + "_insert.txt";
			BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(insertFilePath), Charset.defaultCharset()));
			wrapInsertBufferedWriter.setBufferedWriter(bufferedWriter);

		}
		wrapInsertBufferedWriter.setInsertNumber(wrapInsertBufferedWriter.getInsertNumber() + 1);
		this.writeCustomersToFile(customerDBVO, wrapInsertBufferedWriter.getBufferedWriter());

	}

	void writeCustomerDBVOToUpdatePrepareFile(CustomerDBVO customerDBVO, String jobId, String listId, Set<String> actionSetIds,
			WrapBufferedWriter wrapUpdateBufferedWriter) throws IOException {
		customerDBVO.setUpdateTime(new Date());
		if (StringUtils.isEmpty(customerDBVO.getCustomerListId())) {
			customerDBVO.setCustomerListId(listId);
		}
		customerDBVO.getActionSetIds().addAll(actionSetIds);
		if (wrapUpdateBufferedWriter.getBufferedWriter() == null) {
			String updateDBFilePath = resourcePath + InternalConstants.UPLOAD_CUSTOMER_PREPARE_BG_UPDATE + jobId + "_update.txt";
			BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(updateDBFilePath), Charset.defaultCharset()));
			wrapUpdateBufferedWriter.setBufferedWriter(bufferedWriter);

		}
		wrapUpdateBufferedWriter.setUpdateNumber(wrapUpdateBufferedWriter.getUpdateNumber() + 1);
		this.writeCustomersToFile(customerDBVO, wrapUpdateBufferedWriter.getBufferedWriter());
	}

	CustomerDBVO convertColumnValueToCustomerDBVO(CustomerUploadVO context, Set<CustomerPropertyDBVO> properties, String[] columnValues) throws IOException {
		CustomerDBVO customerDBVO = new CustomerDBVO();
		for (Integer csvColumnIndex : context.getMatchColumn().keySet()) {
			CustomerMappingColumnVO itemSuffixType = context.getMatchColumn().get(csvColumnIndex);
			CustomerColumnType columnType = itemSuffixType.getColumnType();
			String columnValue = null;
			if (csvColumnIndex < columnValues.length) {
				columnValue = columnValues[csvColumnIndex];
			}
			if (StringUtils.isNotEmpty(columnValue)) {
				if (csvColumnIndex == context.getEmailColumnIndex()) {
					String emailAddress = columnValue.toLowerCase();
					EmailSubscriberDBVO emailSubscriberDBVO = new EmailSubscriberDBVO();
					emailSubscriberDBVO.setEmail(emailAddress);
					customerDBVO.getEmailSubscribers().add(emailSubscriberDBVO);
				} else if (csvColumnIndex == context.getMobileColumnIndex()) {
					String mobileAddress = columnValue;
					if (StringUtils.isNotBlank(mobileAddress)) {
						try {
							PhoneNumber phoneNumber = PhoneNumberUtil.parse(mobileAddress);
							MobileSubscriberDBVO mobileSubscriberDBVO = new MobileSubscriberDBVO();
							mobileSubscriberDBVO.setMobileSubscriberId(ObjectIdGenerator.generate());
							mobileSubscriberDBVO.setMobile(String.valueOf(phoneNumber.getNumber()));
							mobileSubscriberDBVO.setCountryCode(phoneNumber.getCountryCode());
							customerDBVO.getMobileSubscribers().add(mobileSubscriberDBVO);
						} catch (InvalidPhoneNumberException e) {
							logger.error("convertColumnValueToCustomerDBVO Caught exception:" + ExceptionUtils.getStackTrace(e));
						}
					}
				} else if (csvColumnIndex == context.getFirstNameColumnIndex()) {
					customerDBVO.setFirstName(columnValue);
				} else if (csvColumnIndex == context.getLastNameColumnIndex()) {
					customerDBVO.setLastName(columnValue);
				} else if (CustomerColumnType.DOUBLE == columnType) {
					Double dou = DoubleValidator.getInstance().validate(columnValue);
					if (dou != null) {
						customerDBVO.getProperties().put(itemSuffixType.getPropertyId(), dou.doubleValue());
					}
				} else if (CustomerColumnType.DATE == columnType) {
					String[] dateFormatArray = new String[] { "yyyy-MM-dd", "MM/dd/yyyy", "yyyy/MM/dd" };
					if (StringUtils.isNotBlank(customerDateFormat)) {
						dateFormatArray = StringUtils.split(customerDateFormat, ";");
					}
					try {
						Date dateValue = DateUtils.parseDateStrictly(columnValue, dateFormatArray);
						customerDBVO.getProperties().put(itemSuffixType.getPropertyId(), dateValue);
					} catch (ParseException e) {
						logger.error("convertColumnValueToCustomerDBVO Caught exception:" + ExceptionUtils.getStackTrace(e));
					}
				} else if (CustomerColumnType.SET == columnType) {
					CustomerPropertyDBVO propDBVO = getPropertyById(itemSuffixType.getPropertyId(), properties);
					Set<CustomerPropertyItemDBVO> itemDBVOs = propDBVO.getItems();
					CustomerPropertyItemDBVO matchedItem = findMatchedPropertyItem(columnValue, itemDBVOs);
					if (matchedItem != null) {
						customerDBVO.getProperties().put(itemSuffixType.getPropertyId(), new String[] { matchedItem.getPropertyItemId() });
					}
				} else if (CustomerColumnType.MULTISET == columnType) {
					CustomerPropertyDBVO propDBVO = getPropertyById(itemSuffixType.getPropertyId(), properties);
					Set<CustomerPropertyItemDBVO> itemDBVOs = propDBVO.getItems();
					String[] itemValues = columnValue.split(SET_SEPARATOR);
					List<String> matchedItems = new ArrayList<String>();
					if (itemValues != null) {
						for (String itemValue : itemValues) {
							CustomerPropertyItemDBVO matchedItem = findMatchedPropertyItem(StringUtils.strip(itemValue), itemDBVOs);
							matchedItems.add(matchedItem.getPropertyItemId());
						}
					}
					customerDBVO.getProperties().put(itemSuffixType.getPropertyId(), matchedItems);
				} else if (csvColumnIndex == context.getManggisIdColumnIndex()) {
					customerDBVO.setId(columnValue);
				} else if (csvColumnIndex == context.getExternalIdColumnIndex()) {
					customerDBVO.setExternalId(columnValue);
				} else {
					customerDBVO.getProperties().put(itemSuffixType.getPropertyId(), columnValue);
				}
			}
		}
		return customerDBVO;
	}

	private CustomerPropertyItemDBVO findMatchedPropertyItem(String item, Set<CustomerPropertyItemDBVO> items) {
		for (CustomerPropertyItemDBVO itemDBVO : items) {
			if (itemDBVO.getValue().equalsIgnoreCase(item)) {
				return itemDBVO;
			}
		}
		return null;
	}

	void writeCustomersToFile(CustomerDBVO customer, BufferedWriter bufferedWriter) throws IOException {
		String jsonString = dbObjectConverter.parseObjectToDBJson(customer);
		bufferedWriter.write(jsonString);
		bufferedWriter.newLine();
	}

	private boolean validateIssue(CustomerUploadVO context, CustomerUploadResultVO resultContext, String[] columnValues, Set<CustomerPropertyDBVO> properties,
			String jobId, String[] generateCsvHeader, String listId, WrapWriter fieldProblemWrapWriter, WrapWriter manggisIdNotFoundWrapWriter,
			WrapWriter firstNameLastNameEmptyWrapWriter) throws IOException {
		boolean existFieldProblem = this.handleInvalidValuesProblem(context, resultContext, columnValues, properties, fieldProblemWrapWriter, jobId,
				generateCsvHeader);
		if (existFieldProblem) {
			return true;
		}
		String manggisId = null;
		String extId = null;
		String email = null;
		String mobile = null;
		String firstName = null;
		String lastName = null;
		for (Integer columnIndex : context.getMatchColumn().keySet()) {
			String columnValue = null;
			if (columnIndex < columnValues.length) {
				columnValue = StringUtils.strip(columnValues[columnIndex]);
			} else {
				continue;
			}
			if (columnIndex == context.getManggisIdColumnIndex()) {
				if (StringUtils.isNotEmpty(columnValue)) {
					manggisId = columnValue;
				}
			} else if (columnIndex == context.getExternalIdColumnIndex()) {
				if (StringUtils.isNotEmpty(columnValue)) {
					extId = columnValue;
				}
			} else if (columnIndex == context.getEmailColumnIndex()) {
				if (StringUtils.isNotEmpty(columnValue)) {
					email = columnValue.toLowerCase();
				}
			} else if (columnIndex == context.getMobileColumnIndex()) {
				if (StringUtils.isNotEmpty(columnValue)) {
					PhoneNumber phoneNumber;
					try {
						phoneNumber = PhoneNumberUtil.parse(columnValue);
						mobile = phoneNumber.getNumber() + "";
					} catch (InvalidPhoneNumberException e) {
						logger.error(e.getMessage());
					}
				}
			} else if (columnIndex == context.getFirstNameColumnIndex()) {
				firstName = columnValue;
			} else if (columnIndex == context.getLastNameColumnIndex()) {
				lastName = columnValue;
			}
		}

		boolean manggisIdNotFound = false;
		if (manggisId != null) {
			manggisIdNotFound = handleNotExistManggisIdProblem(resultContext, columnValues, listId, manggisId, manggisIdNotFoundWrapWriter, jobId,
					generateCsvHeader);
		}
		if (manggisIdNotFound) {
			return true;
		}
		boolean firstNameAndLastNameEmpty = this.handleFirstNameAndLastNameEmptyProblem(resultContext, columnValues, manggisId, extId, email, mobile,
				firstName, lastName, firstNameLastNameEmptyWrapWriter, jobId, generateCsvHeader);
		if (firstNameAndLastNameEmpty) {
			return true;
		}

		return false;
	}

	/**
	 * 检查文件里的数据正确性, Email, Mobile, (Invalid Date||Invalid
	 * Number||Invalid Set||Invalid Multiple Set) 当任何列的数据校验出错.
	 * 剩下列的校验不需要再做.同时这行数据记录在文件中 有任何列的数据有问题时返回true, 否则为false
	 * 
	 * @param context
	 * @param fileReader
	 * @param properties
	 * @param mappingColumns
	 * @return
	 * @throws IOException
	 */
	private boolean handleInvalidValuesProblem(CustomerUploadVO context, CustomerUploadResultVO resultContext, String[] columnValues,
			Set<CustomerPropertyDBVO> properties, WrapWriter fieldProblemWrapWriter, String jobId, String[] generateCsvHeader) throws IOException {
		boolean hasFieldProblem = false;
		for (Integer csvColumnIndex : context.getMatchColumn().keySet()) {
			if (hasFieldProblem) {
				break;
			}
			CustomerMappingColumnVO itemSuffixType = context.getMatchColumn().get(csvColumnIndex);
			CustomerColumnType columnType = itemSuffixType.getColumnType();
			String columnValue = null;
			if (csvColumnIndex < columnValues.length) {
				columnValue = columnValues[csvColumnIndex];
			} else {
				continue;
			}
			if (StringUtils.isNotBlank(columnValue)) {
				String trimColumnValue = StringUtils.strip(columnValue);
				if (csvColumnIndex == context.getEmailColumnIndex()) {
					String emailAddress = trimColumnValue.toLowerCase();
					boolean emailValid = true;
					emailValid = EmailUtil.isValid(emailAddress);
					if (!emailValid) {
						hasFieldProblem = true;
					}
				} else if (csvColumnIndex == context.getMobileColumnIndex()) {
					String mobileAddress = trimColumnValue;
					boolean mobileValid = true;
					if (StringUtils.isNotBlank(mobileAddress)) {
						try {
							PhoneNumberUtil.parse(mobileAddress);
						} catch (InvalidPhoneNumberException e) {
							mobileValid = false;
						}
					}
					if (!mobileValid) {
						hasFieldProblem = true;
					}
				} else if (CustomerColumnType.DOUBLE == columnType) {
					Double dou = DoubleValidator.getInstance().validate(trimColumnValue);
					if (dou == null) {
						hasFieldProblem = true;
					}
				} else if (CustomerColumnType.DATE == columnType) {
					String[] dateFormatArray = new String[] { "yyyy-MM-dd", "MM/dd/yyyy", "yyyy/MM/dd" };
					if (StringUtils.isNotBlank(customerDateFormat)) {
						dateFormatArray = StringUtils.split(customerDateFormat, ";");
					}
					try {
						DateUtils.parseDateStrictly(trimColumnValue, dateFormatArray);
					} catch (ParseException e) {
						hasFieldProblem = true;
					}
				} else if (CustomerColumnType.SET == columnType) {
					CustomerPropertyDBVO propDBVO = getPropertyById(itemSuffixType.getPropertyId(), properties);
					if (propDBVO == null) {
						hasFieldProblem = true;
					} else {
						Set<CustomerPropertyItemDBVO> itemDBVOs = propDBVO.getItems();
						boolean itemExist = checkExistPropertyItem(trimColumnValue, itemDBVOs);
						if (!itemExist) {
							hasFieldProblem = true;
						}
					}
				} else if (CustomerColumnType.MULTISET == columnType) {
					CustomerPropertyDBVO propDBVO = getPropertyById(itemSuffixType.getPropertyId(), properties);
					if (propDBVO == null) {
						hasFieldProblem = true;
					} else {
						Set<CustomerPropertyItemDBVO> itemDBVOs = propDBVO.getItems();
						String[] itemValues = trimColumnValue.split(SET_SEPARATOR);
						if (itemValues != null) {
							for (String itemValue : itemValues) {
								boolean itemExist = checkExistPropertyItem(StringUtils.strip(itemValue), itemDBVOs);
								if (!itemExist) {
									hasFieldProblem = true;
									break;
								}
							}
						}
					}
				}
			}
		}
		if (hasFieldProblem) {
			if (fieldProblemWrapWriter.getCsvWriter() == null) {
				String fieldProblemDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_FIELD_PROBLEM;
				this.createDirectoryNotExist(fieldProblemDirectory);
				String finalFieldProblemFileName = jobId + "_fieldProblem.csv";
				String fieldProblemFilePath = fieldProblemDirectory + finalFieldProblemFileName;
				CsvWriter csvWriter = new CsvWriter(fieldProblemFilePath, Charset.defaultCharset());
				fieldProblemWrapWriter.setCsvWriter(csvWriter);
				csvWriter.writeRecord(generateCsvHeader);
				resultContext.setFieldProblemCsvPath(finalFieldProblemFileName);
			}
			fieldProblemWrapWriter.getCsvWriter().writeRecord(columnValues);
			resultContext.setFieldProblemNumber(resultContext.getFieldProblemNumber() + 1);
		}
		return hasFieldProblem;
	}

	/**
	 * manggisId 找不到就return true, 否则return false
	 * @param fileReader
	 * @param jedis
	 * @param customerListId
	 * @param manggisId
	 * @param manggisIdNotFoundWrapWriter
	 * @param jobId
	 * @param generateCsvHeader
	 * @return
	 * @throws IOException 
	 * @throws Exception
	 */
	private boolean handleNotExistManggisIdProblem(CustomerUploadResultVO resultContext, String[] columnValues, String customerListId, String manggisId,
			WrapWriter manggisIdNotFoundWrapWriter, String jobId, String[] generateCsvHeader) throws IOException {
		Jedis jedis = jedisPool.getResource();
		boolean manggisIdNotFound = false;
		String currentListCustomerIdsKey = customerListId + RedisKeyName.CUSTOMER_IDS;
		boolean existManggisId = false;
		try {
			existManggisId = jedis.sismember(currentListCustomerIdsKey, manggisId);
		} finally {
			jedisPool.returnResource(jedis);
		}
		if (!existManggisId) {
			if (manggisIdNotFoundWrapWriter.getCsvWriter() == null) {
				String manggisIdNotFoundDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_MANGGIS_ID_NOT_FOUND_PROBLEM;
				this.createDirectoryNotExist(manggisIdNotFoundDirectory);
				String manggisIdNotFoundFileName = jobId + "_manggisIdNotFound.csv";
				String manggisIdNotFoundFilePath = manggisIdNotFoundDirectory + manggisIdNotFoundFileName;
				CsvWriter csvWriter = new CsvWriter(manggisIdNotFoundFilePath, Charset.defaultCharset());
				csvWriter.writeRecord(generateCsvHeader);
				manggisIdNotFoundWrapWriter.setCsvWriter(csvWriter);
				resultContext.setManggisIdNotFoundCsvPath(manggisIdNotFoundFileName);
			}
			manggisIdNotFoundWrapWriter.getCsvWriter().writeRecord(columnValues);
			manggisIdNotFound = true;
			resultContext.setManggisIdNotFoundNumber(resultContext.getManggisIdNotFoundNumber() + 1);
		}
		return manggisIdNotFound;
	}

	private boolean checkExistPropertyItem(String item, Set<CustomerPropertyItemDBVO> items) {
		boolean result = false;
		for (CustomerPropertyItemDBVO itemDBVO : items) {
			if (itemDBVO.getValue().equalsIgnoreCase(item)) {
				result = true;
				break;
			}
		}
		return result;
	}

	private void createDirectoryNotExist(String directoryPath) {
		File directoryFile = new File(directoryPath);
		if (!directoryFile.exists()) {
			directoryFile.mkdirs();
		}
	}

	private boolean handleFirstNameAndLastNameEmptyProblem(CustomerUploadResultVO resultContext, String[] columnValues, String manggisId, String extId,
			String email, String mobile, String firstName, String lastName, WrapWriter firstNameLastNameEmptyWrapWriter, String jobId,
			String[] generateCsvHeader) throws IOException {
		boolean existProblem = false;
		if (manggisId == null && extId == null && email == null && mobile == null) {
			if (firstName == null || lastName == null) {
				existProblem = true;
			}
		}

		if (existProblem) {
			if (firstNameLastNameEmptyWrapWriter.getCsvWriter() == null) {
				String firstNameLastNameEmptyDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_FIRSTNAME_LASTNAME_PROBLEM;
				this.createDirectoryNotExist(firstNameLastNameEmptyDirectory);
				String finalFNameAndLNameEmptyProblemFileName = jobId + "_firstNameLastNameEmpty.csv";
				String firstNameLastNameEmptyFilePath = firstNameLastNameEmptyDirectory + finalFNameAndLNameEmptyProblemFileName;
				CsvWriter csvWriter = new CsvWriter(firstNameLastNameEmptyFilePath, Charset.defaultCharset());
				firstNameLastNameEmptyWrapWriter.setCsvWriter(csvWriter);
				csvWriter.writeRecord(generateCsvHeader);
				resultContext.setFirstNameLastNameEmptyCsvPath(finalFNameAndLNameEmptyProblemFileName);
			}
			firstNameLastNameEmptyWrapWriter.getCsvWriter().writeRecord(columnValues);
			resultContext.setFirstNameLastNameEmptyNumber(resultContext.getFirstNameLastNameEmptyNumber() + 1);
		}
		return existProblem;
	}

	CustomerPropertyDBVO getPropertyById(String propertyId, Set<CustomerPropertyDBVO> properties) {
		for (CustomerPropertyDBVO property : properties) {
			if (property.getPropertyId().equals(propertyId)) {
				return property;
			}
		}
		return null;
	}

	void initNecessaryDirectory(String jobId) {

		String uploadCustomerDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER;
		createDirectoryNotExist(uploadCustomerDirectory);
		String insertJobFolder = resourcePath + InternalConstants.UPLOAD_CUSTOMER_BG_INSERT + File.separator + jobId;
		createDirectoryNotExist(insertJobFolder);
		String updateJobFolder = resourcePath + InternalConstants.UPLOAD_CUSTOMER_BG_UPDATE + File.separator + jobId;
		createDirectoryNotExist(updateJobFolder);
		String uploadCustomerSkipDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_SKIP;
		createDirectoryNotExist(uploadCustomerSkipDirectory);
		String prepareInsertDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_PREPARE_BG_INSERT;
		createDirectoryNotExist(prepareInsertDirectory);
		String prepareUpdateDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_PREPARE_BG_UPDATE;
		createDirectoryNotExist(prepareUpdateDirectory);
		String multipleEmailProblemDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_MULTI_EMAIL_PROBLEM;
		createDirectoryNotExist(multipleEmailProblemDirectory);
		String multipleMobileProblemDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER_MULTI_MOBILE_PROBLEM;
		createDirectoryNotExist(multipleMobileProblemDirectory);
	}

	void writeColumnValueToCsvFileForSkipAction(WrapWriter wrapWriter, String writeFilePath, String[] columnValues, String[] generateCsvHeader)
			throws IOException {
		if (wrapWriter.getCsvWriter() == null) {
			CsvWriter csvWriter = new CsvWriter(writeFilePath, Charset.defaultCharset());
			csvWriter.writeRecord(generateCsvHeader);
			wrapWriter.setCsvWriter(csvWriter);
		}
		wrapWriter.getCsvWriter().writeRecord(columnValues);
		wrapWriter.setSkipNum(wrapWriter.getSkipNum() + 1);
	}

	// 把不同的几个文件按照key排序,然后删除原来的文件
	void sortFile(String jobId, final CustomerUploadVO context, String[] generateCsvHeader) throws IOException {
		String uploadCustomerDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER;

		String manggisIdNotNullPath = uploadCustomerDirectory + jobId + MANGGISID_NOT_NULL_FILE_NAME_SUFFIX;
		String extIdNotNullPath = uploadCustomerDirectory + jobId + EXTID_NOT_NULL_FILE_NAME_SUFFIX;
		String emailNotNullPath = uploadCustomerDirectory + jobId + EMAIL_NOT_NULL_FILE_NAME_SUFFIX;
		String mobileNotNullPath = uploadCustomerDirectory + jobId + MOBILE_NOT_NULL_FILE_NAME_SUFFIX;

		String sortManggisIdNotNullPath = uploadCustomerDirectory + jobId + SORT_MANGGISID_NOT_NULL_FILE_NAME_SUFFIX;
		String sortExtIdNotNullPath = uploadCustomerDirectory + jobId + SORT_EXTID_NOT_NULL_FILE_NAME_SUFFIX;
		String sortEmailNotNullPath = uploadCustomerDirectory + jobId + SORT_EMAIL_NOT_NULL_FILE_NAME_SUFFIX;
		String sortMobileNotNullPath = uploadCustomerDirectory + jobId + SORT_MOBILE_NOT_NULL_FILE_NAME_SUFFIX;

		this.sortFileHelper(manggisIdNotNullPath, context.getManggisIdColumnIndex(), sortManggisIdNotNullPath, generateCsvHeader);
		FileUtils.deleteQuietly(new File(manggisIdNotNullPath));
		this.sortFileHelper(extIdNotNullPath, context.getExternalIdColumnIndex(), sortExtIdNotNullPath, generateCsvHeader);
		FileUtils.deleteQuietly(new File(extIdNotNullPath));
		this.sortFileHelper(emailNotNullPath, context.getEmailColumnIndex(), sortEmailNotNullPath, generateCsvHeader);
		FileUtils.deleteQuietly(new File(emailNotNullPath));
		this.sortFileHelper(mobileNotNullPath, context.getMobileColumnIndex(), sortMobileNotNullPath, generateCsvHeader);
		FileUtils.deleteQuietly(new File(mobileNotNullPath));

	}

	private void sortFileHelper(String sourcePath, final int compareIndex, String targetPath, String[] generateCsvHeader) throws IOException {
		// sort manggisId
		File sourceFile = new File(sourcePath);
		if (sourceFile.exists()) {
			Comparator<String> cmp = new Comparator<String>() {
				@Override
				public int compare(String o1, String o2) {
					String[] o1Values = o1.split(",");
					String[] o2Values = o2.split(",");
					return o1Values[compareIndex].compareTo(o2Values[compareIndex]);
				}
			};
			String directory = resourcePath + InternalConstants.UPLOAD_CUSTOMER;
			File directoryFile = new File(directory);
			// omit the first line, which is the header..
			List<File> listOfFiles;
			try {
				listOfFiles = ExternalSort.sortInBatch(sourceFile, cmp, 1024, Charset.defaultCharset(), directoryFile, false, 1, false);
				CsvWriter sortManggisIdNotNullWriter = new CsvWriter(targetPath, Charset.defaultCharset());
				sortManggisIdNotNullWriter.writeRecord(generateCsvHeader);
				sortManggisIdNotNullWriter.close();
				File sortManggisIdNotNullFile = new File(targetPath);
				// now merge with append
				ExternalSort.mergeSortedFiles(listOfFiles, sortManggisIdNotNullFile, cmp, Charset.defaultCharset(), false, true, false);
			} catch (IOException e) {
				e.printStackTrace();
				logger.error(e);
				throw e;
			}
		}
	}

	void createCompleteNotificationEmail(CustomerUploadVO ctx) {
		Set<String> notifyEmails = ctx.getNotifyEmails();
		notifyEmails.add(ctx.getOperator());
		List<EmailNotificationDBVO> emailDBVOList = new ArrayList<EmailNotificationDBVO>();
		for (String notifyEmail : notifyEmails) {
			EmailNotificationDBVO emailDBVO = new EmailNotificationDBVO();
			emailDBVO.setSubject("subject_upload_subscribers.ftl");
			emailDBVO.setTemplate("html_upload_subscribers.ftl");
			emailDBVO.setRecipient(notifyEmail);
			emailDBVO.getData().put("user", ctx.getOperator());
			emailDBVOList.add(emailDBVO);
		}
		if (!emailDBVOList.isEmpty()) {
			emailNotificationAutoRepo.save(emailDBVOList);
		}

	}

	void deleteSortFiles(String jobId) {
		String uploadCustomerDirectory = resourcePath + InternalConstants.UPLOAD_CUSTOMER;
		String sortManggisIdNotNullPath = uploadCustomerDirectory + jobId + CustomerUploadHelper.SORT_MANGGISID_NOT_NULL_FILE_NAME_SUFFIX;
		String sortExtIdNotNullPath = uploadCustomerDirectory + jobId + CustomerUploadHelper.SORT_EXTID_NOT_NULL_FILE_NAME_SUFFIX;
		String sortEmailNotNullPath = uploadCustomerDirectory + jobId + CustomerUploadHelper.SORT_EMAIL_NOT_NULL_FILE_NAME_SUFFIX;
		String sortMobileNotNullPath = uploadCustomerDirectory + jobId + CustomerUploadHelper.SORT_MOBILE_NOT_NULL_FILE_NAME_SUFFIX;

		FileUtils.deleteQuietly(new File(sortManggisIdNotNullPath));
		FileUtils.deleteQuietly(new File(sortExtIdNotNullPath));
		FileUtils.deleteQuietly(new File(sortEmailNotNullPath));
		FileUtils.deleteQuietly(new File(sortMobileNotNullPath));
	}
	// 把在prepare folder里的insert和update的txt文件move到BG文件里, 然后删除prepare folder里的文件
	void movePrepareInsertFileToBGFolder(String jobId) throws IOException {
		String insertPrepareFilePath = resourcePath + InternalConstants.UPLOAD_CUSTOMER_PREPARE_BG_INSERT + jobId + "_insert.txt";
		File insertPrepareFile = new File(insertPrepareFilePath);
		String insertBGJobFolder = resourcePath + InternalConstants.UPLOAD_CUSTOMER_BG_INSERT + jobId;
		FileUtils.copyFileToDirectory(insertPrepareFile, new File(insertBGJobFolder));
		insertPrepareFile.delete();
	}

	void movePrepareUpdateFileToBGFolder(String jobId) throws IOException {
		String updatePrepareFilePath = resourcePath + InternalConstants.UPLOAD_CUSTOMER_PREPARE_BG_UPDATE + jobId + "_update.txt";
		File updatePrepareFile = new File(updatePrepareFilePath);
		String updateBGJobFolder = resourcePath + InternalConstants.UPLOAD_CUSTOMER_BG_UPDATE + jobId;
		FileUtils.copyFileToDirectory(updatePrepareFile, new File(updateBGJobFolder));
		updatePrepareFile.delete();
	}

	String loadPrimaryEmail(Set<EmailSubscriberDBVO> emailSubscribers) {
		String email = null;
		for (EmailSubscriberDBVO item : emailSubscribers) {
			if (item.isPrimary()) {
				email = item.getEmail();
			}
		}
		return email;
	}

	String loadPrimaryMobile(Set<MobileSubscriberDBVO> mobileSubscribers) {
		String mobile = null;
		for (MobileSubscriberDBVO item : mobileSubscribers) {
			if (item.isPrimary()) {
				mobile = item.getMobile();
			}
		}
		return mobile;
	}

	void resetPrimaryEmail(Set<EmailSubscriberDBVO> emailSubscribers, String email) {
		for (EmailSubscriberDBVO item : emailSubscribers) {
			if (item.getEmail().equals(email)) {
				item.setPrimary(true);
			} else {
				item.setPrimary(false);
			}
		}
	}

	void resetPrimaryMobile(Set<MobileSubscriberDBVO> mobileSubscribers, String mobile) {
		for (MobileSubscriberDBVO item : mobileSubscribers) {
			if (item.getMobile().equals(mobile)) {
				item.setPrimary(true);
			} else {
				item.setPrimary(false);
			}
		}
	}

	void jedisIncrByCount(Jedis jedis, String jobId, long lineIndex) {
		if (lineIndex % batchSize_updateRedis == 0) {
			jedis.incrBy(jobId + RedisKeyName.CUSTOMER_HANDLE_COUNT, batchSize_updateRedis);
		}
	}

	void jedisIncrByRemain(Jedis jedis, String jobId, long lineIndex) {
		if (lineIndex % batchSize_updateRedis != 0) {
			jedis.incrBy(jobId + RedisKeyName.CUSTOMER_HANDLE_COUNT, (lineIndex % batchSize_updateRedis));
		}
	}

	String populdatedSortFileNotNullPath(String jobId, String sortFileNotNullFileNameSuffix) {
		return resourcePath + InternalConstants.UPLOAD_CUSTOMER + jobId + sortFileNotNullFileNameSuffix;
	}

	String retrieveSkipFilePath(String jobId) {
		return resourcePath + InternalConstants.UPLOAD_CUSTOMER_SKIP + jobId + "_skip.csv";
	}

	String retrieveMultipleEmailFilePath(String jobId) {
		return resourcePath + InternalConstants.UPLOAD_CUSTOMER_MULTI_EMAIL_PROBLEM + jobId + "_multipleEmail.csv";
	}

	String retrieveMultipleMobileFilePath(String jobId) {
		return resourcePath + InternalConstants.UPLOAD_CUSTOMER_MULTI_MOBILE_PROBLEM + jobId + "_multipleMobile.csv";
	}

}
